package com.aaron.remoting.transport.netty.server;

import com.aaron.enums.CompressTypeEnum;
import com.aaron.enums.RpcResponseCodeEnum;
import com.aaron.enums.SerializationTypeEnum;
import com.aaron.factory.SingletonFactory;
import com.aaron.remoting.constants.RpcConstants;
import com.aaron.remoting.dto.RpcMessage;
import com.aaron.remoting.dto.RpcRequest;
import com.aaron.remoting.dto.RpcResponse;
import com.aaron.remoting.handler.RpcRequestHandler;

import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * 如果继承自 SimpleChannelInboundHandler 的话就不要考虑 ByteBuf 的释放 ，内部的
 * channelRead 方法会替你释放 ByteBuf ，避免可能导致的内存泄露问题。
 */
@Slf4j
public class NettyRpcServerHandler extends ChannelInboundHandlerAdapter {

    private final RpcRequestHandler rpcRequestHandler;

    public NettyRpcServerHandler() {
        this.rpcRequestHandler = SingletonFactory.getInstance(RpcRequestHandler.class);
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            if (msg instanceof RpcMessage) {
                log.info("server receive msg: [{}] ", msg);
                // 开始填充回复给客户端的RpcMessage
                byte messageType = ((RpcMessage) msg).getMessageType();
                RpcMessage rpcMessage = new RpcMessage();
                rpcMessage.setCodec(SerializationTypeEnum.PROTOSTUFF.getCode());
                rpcMessage.setCompress(CompressTypeEnum.GZIP.getCode());
                // 如果这个消息只是ping,那么就填充pong
                if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
                    rpcMessage.setMessageType(RpcConstants.HEARTBEAT_RESPONSE_TYPE);
                    rpcMessage.setData(RpcConstants.PONG);
                } else {
                    RpcRequest rpcRequest = (RpcRequest) ((RpcMessage) msg).getData();
                    // 注意，这里是执行目标方法（客户端需要执行的方法）并返回方法结果。服务端使用反射去执行的。
                    Object result = rpcRequestHandler.handle(rpcRequest);
                    // 服务端计算后得到的结果是：
                    log.info(String.format("server get result: %s", result.toString()));
                    // RpcMessage的type设置为2，表示这是回复消息Response类型
                    rpcMessage.setMessageType(RpcConstants.RESPONSE_TYPE);
                    // 管道是活动且可写入的
                    if (ctx.channel().isActive() && ctx.channel().isWritable()) {
                        // 远程调用成功，response对象中继续填入成功的一些信息
                        RpcResponse<Object> rpcResponse = RpcResponse.success(result, rpcRequest.getRequestId());
                        rpcMessage.setData(rpcResponse);
                    } else {
                        // 远程调用失败，response对象中继续填入失败的一些信息
                        RpcResponse<Object> rpcResponse = RpcResponse.fail(RpcResponseCodeEnum.FAIL);
                        rpcMessage.setData(rpcResponse);
                        log.error("not writable now, message dropped");
                    }
                }
                // 发送rpcMessage信息到远端，如果没有回应就关闭连接。这里也是在出站，调用encode()方法？
                ctx.writeAndFlush(rpcMessage).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            }
        } finally {
            // 手动确保释放字节Buf，否则可能有内存泄漏
            ReferenceCountUtil.release(msg);
        }
    }

    // netty的心跳机制。保证客户端和服务端的连接不被断掉，避免重连
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleState state = ((IdleStateEvent) evt).state();
            if (state == IdleState.READER_IDLE) {
                log.info("idle check happen, so close the connection");
                ctx.close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        log.error("server catch exception");
        cause.printStackTrace();
        ctx.close();
    }
}
